package com.xiaoxu.spark.DateFrameWithDataset

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}

/**
  * 通过反射实现RDD --> DataFrame
  */
object DataFrameWithRDD_Demo {

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession
      .builder()
      .appName("DataFrameWithRDD_Demo")
      .master("local[2]")
      .getOrCreate()

    peopleReflection(sparkSession)

    //sparkSession.stop()
    progrom(sparkSession)

  }

  private def peopleReflection(sparkSession: SparkSession) = {
    val rdd = sparkSession.sparkContext.textFile("D:/test/data/people.txt")

    //todo 需要导入隐式转换
    import sparkSession.implicits._
    //RDD --> DataFrame  //1,xiaowang,15 //[1,xiaowang,15]
    val peopleDf = rdd
      .map(_.split(","))
      .map(line => people(line(0).toInt, line(1), line(2).toInt))
      .toDF()

    peopleDf.show()

    peopleDf.filter(
      peopleDf.col("age") > 17
    ).show()

    //--把DataFrame转换为临时表,使用SQL编程方式
    peopleDf.createOrReplaceTempView("peoples")

    sparkSession
      .sql("select * from peoples where age > 17")
      .show()
  }


  def progrom(sparkSession: SparkSession): Unit ={
    val rdd = sparkSession
      .sparkContext
      .textFile("D:/test/data/people.txt")

    val peopleRDD = rdd
      .map(_.split(","))
      .map(line => Row(line(0).toInt,line(1),line(2).toInt))

    val fieldArray = Array(
      StructField("id",IntegerType,true),
      StructField("name",StringType,true),
      StructField("age",IntegerType,true)
    )

    val structType = StructType(fieldArray)

    val peopleDataFrame = sparkSession.createDataFrame(peopleRDD,structType)

    peopleDataFrame.printSchema()
    peopleDataFrame.show()


    //通过df的api进行操作
    peopleDataFrame.filter(peopleDataFrame.col("age") > 30).show

    //通过sql的方式进行操作
    peopleDataFrame.createOrReplaceTempView("people")
    sparkSession.sql("select * from people where age > 17").show()


  }

  case class people(id: Int, name: String, age: Int)

}
